package cn.tedu.mall.order.consumer;

import cn.tedu.mall.order.service.IOrderAddService;
import cn.tedu.mall.pojo.order.dto.OrderAddCondition;
import cn.tedu.mall.pojo.order.dto.OrderAddDTO;
import cn.tedu.mall.pojo.seckill.dto.SeckillOrderAddDTO;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.Random;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
@RocketMQMessageListener(
        topic = "csmall-order-add-topic",
        consumerGroup = "csmall-order-add-group",
        selectorExpression = "*")
public class OrderAddConsumer implements RocketMQListener<MessageExt> {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private IOrderAddService orderAddService;

    private static final String ORDER_ADD_LOCK = "order:add:lock:";

    @Override
    public void onMessage(MessageExt message) {
        //可以从消息上下文中拿到tag 判断条件
        String tags = message.getTags();
        OrderAddCondition orderAddCondition = null;
        String sn = null;
        //orderAdd的json
        String msgBody = new String(message.getBody());
        if ("localAdd".equals(tags)) {
            //转化对象
            orderAddCondition = JSON.toJavaObject(
                    JSON.parseObject(msgBody), OrderAddDTO.class);
            sn = orderAddCondition.getSn();
        } else if ("seckillAdd".equals(tags)) {
            //转化对象
            orderAddCondition = JSON.toJavaObject(
                    JSON.parseObject(msgBody), SeckillOrderAddDTO.class);
            sn = orderAddCondition.getSn();
        }
        //如果两个if都没满足,当前逻辑停止了,不能继续进行
        if (orderAddCondition == null || sn == null) {
            log.error("当前消息:{},没有消费成功,tags不是选型中内容:{}",
                    message.getMsgId(), tags);
            return;
        }
        //设置一个锁key值
        String orderAddKey = ORDER_ADD_LOCK + sn;
        //生成一个随机数,抢锁
        String randCode = new Random().nextInt(9000) + 1000 + "";
        ValueOperations valueOperations = redisTemplate.opsForValue();
        //业务逻辑 查询订单是否存在 不存在则调用addOrder方法
        try {
            //执行抢锁 SETNX key randCode EX 10
            Boolean getKey = valueOperations.setIfAbsent(
                    orderAddKey, randCode, 10, TimeUnit.SECONDS);
            if (getKey) {//抢成功执行addOrder方法
                //执行新增订单 orderServiceImpl 保证幂等 先查sn存在
                log.debug("消费到对象:{}", orderAddCondition);
                orderAddService.addOrder(orderAddCondition);
            } else {
                return;//抢锁失败了结束
            }
        } catch (Exception e) {
            log.error("出现不明异常:{}", e);
            return;
        } finally {
            String randCodeCache = (String) valueOperations.get(orderAddKey);
            if (randCodeCache != null && randCodeCache.equals(randCode)) {
                redisTemplate.delete(orderAddKey);//释放锁
            }
        }
    }
}
